package com.shujia.spark.opt

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

import java.sql.{Connection, DriverManager, PreparedStatement}

object Demo5ForeachPartition {
  def main(args: Array[String]): Unit = {
    //创建spark sql环境（新版spark统一的入口）
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("cache")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    //获取sparkContext
    val sc: SparkContext = spark.sparkContext

    //读取数据
    val studentRDD: RDD[String] = sc.textFile("data/students.txt")

    /**
     * 将RDD的数据保存到数据库中
     */

    /*  studentRDD.foreach(line => {

        val start: Long = System.currentTimeMillis()
        //1、加载驱动
        Class.forName("com.mysql.jdbc.Driver")
        //2、创建数据库连接
        //jdbc底层是网络，需要建立网络连接，底层是tcp/ip协议(需要三次握手)
        val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=UTF-8", "root", "123456")
        //3、编写sql
        val stat: PreparedStatement = con.prepareStatement("insert into students values(?,?,?,?,?)")
        val end: Long = System.currentTimeMillis()
        println(end - start)


        //4、赋值
        val split: Array[String] = line.split(",")
        val id: String = split(0)
        val name: String = split(1)
        val age: Int = split(2).toInt
        val sex: String = split(3)
        val clazz: String = split(4)

        stat.setString(1, id)
        stat.setString(2, name)
        stat.setInt(3, age)
        stat.setString(4, sex)
        stat.setString(5, clazz)



        //5、执行sql
        stat.execute()

        //6、关闭连接
        con.close()
      })

  */
    /**
     * foreachPartition:一次处理一个分区的数据
     * foreachPartition代替foreach可以减少连接数量，只需要为每一个分区创建一个数据库连接
     */

    studentRDD.foreachPartition((iter: Iterator[String]) => {

      /**
       * 每一个分区创建一个网络连接
       */

      val start: Long = System.currentTimeMillis()
      //1、加载驱动
      Class.forName("com.mysql.jdbc.Driver")
      //2、创建数据库连接
      //jdbc底层是网络，需要建立网络连接，底层是tcp/ip协议(需要三次握手)
      val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=UTF-8", "root", "123456")
      //3、编写sql
      val stat: PreparedStatement = con.prepareStatement("insert into students values(?,?,?,?,?)")
      val end: Long = System.currentTimeMillis()
      println(end - start)

      //循环一个分区内的数据
      iter.foreach(line => {
        //4、赋值
        val split: Array[String] = line.split(",")
        val id: String = split(0)
        val name: String = split(1)
        val age: Int = split(2).toInt
        val sex: String = split(3)
        val clazz: String = split(4)

        stat.setString(1, id)
        stat.setString(2, name)
        stat.setInt(3, age)
        stat.setString(4, sex)
        stat.setString(5, clazz)
        //5、执行sql
        stat.execute()

      })
      con.close()
    })
  }
}
